package com.wxzhou.wangzhuan.repository;

import com.wxzhou.wangzhuan.dto.AccountChangeType;
import com.wxzhou.wangzhuan.dto.UserInfoDto;
import com.wxzhou.wangzhuan.entity.*;
import io.smallrye.mutiny.Uni;
import io.vertx.core.json.JsonObject;
import io.vertx.mutiny.sqlclient.Pool;
import io.vertx.mutiny.sqlclient.Row;
import io.vertx.mutiny.sqlclient.Tuple;
import jakarta.enterprise.context.ApplicationScoped;
import jakarta.inject.Inject;
import org.jboss.logging.Logger;

import java.math.BigDecimal;
import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.util.*;

@ApplicationScoped
public class AdRespository {

    private static final Logger LOG = Logger.getLogger(AdRespository.class);
    @Inject
    Pool client;


    public Uni<AdsenseAndChannel> findAdsenseAndChannelBySense(int sense) {
        String sql = "select distinct a.ad_sense sense, a.id adsenseid, a.ad_id adid,a.ad_type ad_type,m.ad_channel ad_channel,m.channel_name app_channel,wan.id ad_nt from wz_app_adsense a " +
                "join public.wz_app_media m on a.media_id = m.id left join public.wz_ad_network wan on m.ad_channel = wan.ad_channel where a.ad_sense = $1 limit 1";
        return client.preparedQuery(sql).execute(Tuple.of(sense)).onItem().transform(rows -> {
            if (rows.size() > 0) {
                Row row = rows.iterator().next();
                AdsenseAndChannel channel = new AdsenseAndChannel();
                channel.setAd_channel(row.getInteger("ad_channel"));
                channel.setAdsenseid(row.getLong("adsenseid"));
                channel.setSense(row.getInteger("sense"));
                channel.setAdid(row.getString("adid"));
                channel.setApp_channel(row.getString("app_channel"));
                channel.setAd_type(row.getInteger("ad_type"));
                channel.setAd_nt(row.getInteger("ad_nt"));
                return channel;
            }
            return null;
        }).onFailure().invoke(throwable -> throwable.printStackTrace());
    }


    /**
     * 通过上报的渠道唯一标识，查询所有的广告场景。
     *
     * @param appChannelName
     * @return
     */
    public Uni<List<AdsenseAndChannel>> findAllAdsenseByChannel(String appChannelName) {
        String sql = "select distinct a.ad_sense sense, a.id adsenseid, a.ad_id adid,a.ad_type ad_type,m.ad_channel ad_channel,m.channel_name app_channel,wan.id ad_nt " +
                " from wz_app_adsense a join wz_app_media m on a.media_id = m.id left join public.wz_ad_network wan on m.ad_channel = wan.ad_channel where m.channel_name = $1";
        return client.preparedQuery(sql).execute(Tuple.of(appChannelName)).onItem().transform(rows -> {
            List<AdsenseAndChannel> adLogs = new ArrayList<>();
            for (Row row : rows) {
                AdsenseAndChannel channel = new AdsenseAndChannel();
                channel.setAd_channel(row.getInteger("ad_channel"));
                channel.setAdsenseid(row.getLong("adsenseid"));
                channel.setSense(row.getInteger("sense"));
                channel.setAdid(row.getString("adid"));
                channel.setApp_channel(row.getString("app_channel"));
                channel.setAd_type(row.getInteger("ad_type"));
                channel.setAd_nt(row.getInteger("ad_nt"));
                adLogs.add(channel);
            }
            return adLogs;
        }).onFailure().invoke(throwable -> throwable.printStackTrace());

    }

    /**
     * 根据一批adidsns，查找到所有bs_ac=0存在，但结果中不包含的数据
     *
     * @param adidsns
     */
    public Uni<List<AdLog>> findAdidsns(Set<String> adidsns, String deviceid, Long userid, String app_channel) {
        if (adidsns == null || adidsns.isEmpty()) {
            return Uni.createFrom().item(new ArrayList<>());
        }

        // 动态生成IN子句的占位符
        List<String> placeholders = new ArrayList<>();
        List adidsnList = new ArrayList<>(adidsns);

        for (int i = 0; i < adidsnList.size(); i++) {
            placeholders.add("$" + (i + 1));
        }

        String inClause = String.join(", ", placeholders);

        StringBuilder sql = new StringBuilder("SELECT main.* FROM wz_ad_log AS main WHERE main.bs_ac >= 0 AND main.ad_id_sn IN (" + inClause + ") AND main.created_at >= NOW() - INTERVAL '24 HOURS'");

        sql.append(" AND main.deviceid = $").append(placeholders.size() + 1);
        adidsnList.add(deviceid);
        sql.append(" AND main.user_id = $").append(placeholders.size() + 2);
        adidsnList.add(userid);
        sql.append(" AND main.app_channel = $").append(placeholders.size() + 3);
        adidsnList.add(app_channel);

        // 将Set转换为List并传递给Tuple
        return client.preparedQuery(sql.toString()).execute(Tuple.from(adidsnList)).onItem().transform(rows -> {
            List<AdLog> adLogs = new ArrayList<>();
            for (Row row : rows) {
                adLogs.add(mapRowToAdlog(row));
            }
            return adLogs;
        });
    }


    /**
     * 根据app渠道、广告场景，查找对应的广告配置
     */
//    public Uni<AppAdsense> findAdidBy(Integer adsense,String app_channel) {
//        String sql = "select ad.* from wz_app_adsense ad join wz_app_media media on ad.media_id = media.id and media.channel_name = $1 and ad.ad_sense = $2 " +
//                "AND media.channel_name <> '' and ad.ad_id <> '' and media.status = 1 LIMIT 1";
//        return client.preparedQuery(sql)
//                .execute(Tuple.of(app_channel, adsense))
//                .onItem().transform(rows -> {
//                    if (rows.size() > 0) {
//                        return mapRowToAppAdsense(rows.iterator().next());
//                    }
//                    return null;
//                });
//    }

//    /**
//     * 根据广告业务组id，查询广告日志链路
//     *
//     * @param adidsn
//     * @return
//     */
//    public Uni<List<AdLog>> findAllByAdidsn(String adidsn) {
//        if (adidsn == null || adidsn.isBlank()) throw new RuntimeException("adidsn can not be null");
//        String sql = "SELECT * FROM wz_ad_log WHERE ad_id_sn IS NOT NULL AND ad_id_sn <> '' AND ad_id_sn = $1 LIMIT 10";
//        return client.preparedQuery(sql).execute(Tuple.of(adidsn)).onItem().transform(rows -> {
//            List<AdLog> adLogs = new ArrayList<>();
//            for (Row row : rows) {
//                adLogs.add(mapRowToAdlog(row));
//            }
//            return adLogs;
//        });
//    }

    /**
     * 批量插入预请求数据，插入到wz_pre_ad_log表中。
     * @param adLog 预拉请求对象
     * @return
     */
    public Uni<Void> insertPreLog(AdLog adLog) {
        List<String> insertColumns = Arrays.asList("created_at", "deviceid", "app_channel", "ad_qd", "user_id", "bs_ac", "ad_id", "ad_id_sn", "remark", "report_time", "ad_sense", "ad_type");
        // 列名拼接
        String columns = String.join(", ", insertColumns);

        // 动态生成占位符（$1, $2, $3...）
        List<String> placeholderList = new ArrayList<>();
        for (int i = 0; i < insertColumns.size(); i++) {
            placeholderList.add("$" + (i + 1));
        }
        String placeholders = String.join(", ", placeholderList);

        String sql = String.format("INSERT INTO wz_pre_ad_log (%s) VALUES (%s)  RETURNING id", columns, placeholders);

        Tuple tuple = Tuple.from(Arrays.asList(
                LocalDateTime.now(), //插入时间是当前时间
                adLog.getDeviceid(),
                adLog.getApp_channel(),
                adLog.getAd_qd(),
                adLog.getUser_id(),
                adLog.getBs_ac(),
                adLog.getAd_id(),
                adLog.getAd_id_sn(),
                adLog.getRemark(),
                adLog.getReport_time(), //上报时间是
                adLog.getAd_sense(),
                adLog.getAd_type()
        ));

        return client.preparedQuery(sql.toString()).execute(tuple).onItem().transformToUni(rows -> Uni.createFrom().voidItem());
    }

    /**
     * 用户收入增加。
     *
     * @param amount
     * @param adidsns 广告分组编号,用于追溯
     * @param userId
     * @return
     */
    public Uni<Void> saveLogAmountAndReturn(String transId,BigDecimal amount, String adidsns, Long userId,int changeType,String remark) {
        if (amount.compareTo(BigDecimal.ZERO) <= 0) return Uni.createFrom().voidItem();
        //userid为当前收益发生用户，查询其pid
        return client.withTransaction(conn -> {
            String querMainUser = "SELECT u.id uid,u.is_main u_is_main,p.id parentid,p.is_main parent_is_main,re.user_id confirm_user_id FROM wz_users u LEFT JOIN wz_users p ON u.pid = p.id LEFT JOIN wz_jxb_user_bind_relation re ON u.id = re.user_id AND p.id = re.main_user_id WHERE u.pid > 0 AND p.is_main = true AND u.id = $1";

            return conn.preparedQuery(querMainUser).execute(Tuple.of(userId)).onItem().transform(rows -> {
                if (rows.size() <= 0) {
                    LOG.errorf("用户账户不存在，userId: %s", userId);
                    throw new RuntimeException("用户账户不存在，不继续执行");
                }
                return rows.iterator().next(); // 返回账户信息
            }).onItem().transformToUni(currentUserRow -> {
                Long currentUserId = currentUserRow.getLong("uid"); //当前用户
                Long confirm_user_id = currentUserRow.getLong("confirm_user_id"); //当前用户确认绑定了关系!
                Boolean currentUserIsMain = currentUserRow.getBoolean("u_is_main");
                Long parentUserId = currentUserRow.getLong("parentid");
                Boolean parentUserIsMain = currentUserRow.getBoolean("parent_is_main");
                if (currentUserId == null || currentUserId <= 0 || confirm_user_id == null || confirm_user_id <= 0) {
                    throw new RuntimeException("用户账户不存在，不继续执行");
                }
                Long amountForWho;
                if (currentUserIsMain == true) {
                    //直接给当前用户
                    amountForWho = currentUserId;
                } else {
                    if (parentUserId != null && parentUserId > 0 && parentUserIsMain == true) {
                        amountForWho = parentUserId; //给主账号
                    } else {
                        amountForWho = 0L; //不给任何人
                    }
                }
                // 4. 锁定用户账户行
                String lockAccountSql = "SELECT * FROM wz_user_accounts WHERE user_id = $1 FOR UPDATE";
                Tuple lockTuple = Tuple.of(amountForWho);
                return conn.preparedQuery(lockAccountSql).execute(lockTuple).onItem().transform(rows -> {
                    if (rows.size() == 0) {
                        LOG.errorf("用户账户不存在，userId: %s", amountForWho);
                        throw new RuntimeException("用户账户不存在，不继续执行");
                    }
                    return rows.iterator().next(); // 返回账户信息
                }).onItem().transformToUni(accountRow -> {
                    BigDecimal balance = accountRow.getBigDecimal("balance");
                    // 5. 插入账户变更记录
                    String changeSql = "INSERT INTO wz_user_account_changes (user_id, account_id, change_type, change_amount, balance_before, balance_after, related_order_no, remark, created_at,bs_ac,trans_id,sys_remark,from_user_id) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9,$10,$11,$12,$13) RETURNING id";
                    Tuple tuple = Tuple.from(Arrays.asList(
                            accountRow.getLong("user_id"),
                            accountRow.getLong("id"),
                            AccountChangeType.INCOME.getCode(),
                            amount, // 实际以wz_user_withdrawals的amount为准
                            balance, // 变更前余额，单位分
                            balance.add(amount), // 变更后余额
                            "", //没有订单号
                            remark,//备注
                            LocalDateTime.now(),
                            changeType, //账变业务场景
                            transId,
                            adidsns,
                            userId));
                    return conn.preparedQuery(changeSql).execute(tuple).onItem().transformToUni(rows -> {
                        // 6. 修改账户余额，返还金额，余额金额要够
                        String updateAccountSql = "UPDATE wz_user_accounts SET balance = balance + $1  WHERE user_id = $2 RETURNING *";
                        return conn.preparedQuery(updateAccountSql).execute(Tuple.from(Arrays.asList(amount, amountForWho)))
                                .onItem().transformToUni(accountUpdateRows -> {
                                    // 检查账户余额更新是否成功（影响行数大于0）
                                    if (accountUpdateRows.rowCount() <= 0) {
                                        LOG.errorf("广告收入账户余额更新失败，userId: %s, 影响行数: %s", amountForWho, accountUpdateRows.rowCount());
                                        throw new RuntimeException("账户余额更新失败，不继续执行");
                                    }
                                    return Uni.createFrom().voidItem();
                                });
                    });
                }).onFailure().invoke(throwable -> {
                    LOG.errorf(throwable, "插入数据库失败");
                });
            });

        });
    }

    /**
     * 批量插入到广告日志数据库,返回条数
     *
     * @param adLogs
     * @return
     */
    public Uni<Integer> batchInsert(List<AdLog> adLogs) {
        if (adLogs == null || adLogs.isEmpty()) {
            return Uni.createFrom().item(0);
        }

        // 安全检查：确保批量大小在合理范围内
        final int MAX_BATCH_SIZE = 2000; // 最大批量大小
        if (adLogs.size() > MAX_BATCH_SIZE) {
            LOG.errorf("wz_ad_log", "批量登录日志数量过大，建议分批处理", "requestedSize=" + adLogs.size() + ", maxBatchSize=" + MAX_BATCH_SIZE);
            // 这里可以选择抛出异常或者继续处理，为了兼容性选择继续处理
        }

        List<String> insertColumns = Arrays.asList("created_at", "deviceid", "app_channel", "ad_qd", "user_id", "bs_ac", "ad_id", "ad_id_pk", "ad_id_sn", "report_time", "ad_info", "ad_nt", "ad_type", "ad_sense", "ecpm","trans_id","user_amount","butie_amount");

        // 计算SQL参数数量：每条7个字段
        final int FIELDS_PER_COMMENT = insertColumns.size();
        final int totalParams = adLogs.size() * FIELDS_PER_COMMENT;

        // 安全检查：确保参数数量在合理范围内
        final int MAX_PARAMS = 50000; // 最大参数数量限制
        if (totalParams > MAX_PARAMS) {
            LOG.errorf("wz_ad_log", "批量登录日志参数数量过多，可能导致性能问题", "totalParams=" + totalParams + ", maxParams=" + MAX_PARAMS + ",count=" + adLogs.size());
            return Uni.createFrom().failure(new IllegalArgumentException("批量登录日志参数数量过多: " + totalParams + " > " + MAX_PARAMS));
        }

        // 列名拼接
        String columns = String.join(", ", insertColumns);

        // 动态生成占位符（$1, $2, $3...）
        List<String> placeholderList = new ArrayList<>();
        //批量构造插入语句
        int tempindex = 0;
        List tuples = new ArrayList();
        for (int i = 0; i < adLogs.size(); i++) {
            AdLog adLog = adLogs.get(i);
            List<String> tempPlaceholderList = new ArrayList<>();
            for (int j = 0; j < insertColumns.size(); j++) {
                //i=1 ，insertColumns.size()+i开始
                tempindex = j + 1 + i * insertColumns.size();
                tempPlaceholderList.add("$" + tempindex);
            }
            List list = Arrays.asList(adLog.getCreated_at(), //上报时间必须来自客户端时间。
                    adLog.getDeviceid(), adLog.getApp_channel(), adLog.getAd_qd(), adLog.getUser_id(), adLog.getBs_ac(), adLog.getAd_id(),
                    adLog.getAd_id_pk(), adLog.getAd_id_sn(), adLog.getReport_time(), adLog.getAd_info(),
                    adLog.getAd_nt(), adLog.getAd_type(), adLog.getAd_sense(), adLog.getEcpm(),adLog.getTrans_id(),
                    adLog.getUser_amount(),adLog.getButie_amount()
            );
            tuples.addAll(list);
            String format = String.join(", ", tempPlaceholderList);
            String format1 = String.format("(%s)", format);
            placeholderList.add(format1);
        }

        String placeholders = String.join(", ", placeholderList);

        String sql = String.format("INSERT INTO wz_ad_log (%s) VALUES %s ", columns, placeholders);

        return client.preparedQuery(sql.toString()).execute(Tuple.from(tuples)).onItem().transform(rows -> {
            int rowCount = rows.rowCount();
            LOG.infof("wz_ad_log 批量登录日志成功 count %s", rowCount);
            return rowCount;
        }).onFailure().recoverWithUni(throwable -> {
            LOG.errorf(throwable, "wz_ad_log", "批量登录日志失败 count=%s" + (adLogs != null ? adLogs.size() : "0"));
            return Uni.createFrom().item(0);
        });
    }

    private AppAdsense mapRowToAppAdsense(Row row) {
        AppAdsense config = new AppAdsense();
        config.setId(row.getLong("id"));
        config.setAd_name(row.getString("ad_name"));
        config.setAd_type(row.getInteger("ad_type"));
        config.setRemark(row.getString("remark"));
        config.setMedia_id(row.getLong("media_id"));
        config.setAd_id(row.getString("ad_id"));
        config.setAd_sense(row.getInteger("ad_sense"));
        return config;
    }

    private AdLog mapRowToAdlog(Row row) {
        AdLog config = new AdLog();
        config.setId(row.getLong("id"));
        config.setCreated_at(row.getLocalDateTime("created_at"));
        config.setDeviceid(row.getString("deviceid"));
        config.setUser_id(row.getLong("user_id"));
        config.setAd_qd(row.getInteger("ad_qd"));
        config.setAd_nt(row.getInteger("ad_nt"));
        config.setBs_ac(row.getInteger("bs_ac"));
        config.setAd_id(row.getString("ad_id"));
        config.setAd_id_pk(row.getLong("ad_id_pk"));
        config.setAd_id_sn(row.getString("ad_id_sn"));
        config.setEcpm(row.getInteger("ecpm"));
        config.setRate(row.getBigDecimal("rate"));
        config.setUser_amount(row.getBigDecimal("user_amount"));
        config.setButie_amount(row.getBigDecimal("butie_amount"));
        config.setReal_ad_amount(row.getBigDecimal("real_ad_amount"));
        config.setRemark(row.getString("remark"));
        config.setAd_type(row.getInteger("ad_type"));
        config.setAd_sense(row.getInteger("ad_sense"));
        config.setApp_channel(row.getString("app_channel"));
        config.setReport_time(row.getLocalDateTime("report_time"));

        try {
            JsonObject adInfo = row.getJsonObject("ad_info");
            config.setAd_info(adInfo == null ? null : adInfo.toString());
        } catch (Exception e) {
            config.setAd_info("");
        }
        return config;
    }
}
